package cn.itcast.flink.base;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;

/**
 * Author itcast
 * Date 2021/7/27 18:17
 * 从 socket 中读取数据，将 wordcount 之后
 * 保存到 redis 中，使用 hset : hash key:value
 */
public class RedisConnector {
    public static void main(String[] args) throws Exception {
        //开启 nc -lk 9999
        //生成流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //读取socket数据源
        DataStreamSource<String> source = env.socketTextStream("node1", 9999);
        //进行wordcount的计算
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = source.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        }).keyBy(t -> t.f0)
                .sum(1);
        //开启 redis
        //将数据通过connector 保存到redis中
        result.addSink(new RedisSink<Tuple2<String,Integer>>(
                new FlinkJedisPoolConfig.Builder()
                        .setDatabase(10)
                        .setHost("node3")
                        .setPort(6379)
                        .build(),
                new RedisMapper<Tuple2<String,Integer>>() {
                    @Override
                    public RedisCommandDescription getCommandDescription() {
                        return new RedisCommandDescription(
                                RedisCommand.HSET,
                                "wordcount"
                        );
                    }

                    @Override
                    public String getKeyFromData(Tuple2<String,Integer> data) {
                        return data.f0;
                    }

                    @Override
                    public String getValueFromData(Tuple2<String,Integer> data) {
                        return data.f1.toString();
                    }
                }
        ));
        //执行环境
        env.execute();
        //查验结果
    }
}
